import math
import re
from typing import List

import pandas as pd
from pandas import DataFrame
from pandas._typing import IntStrT

from backend.app.database.db_pinecone import rdb_conn
from backend.app.schemas.device_v2 import DeviceSchemaV2
from backend.app.script.data_clean import DataCleanChain


class DataSourceByExcel(object):
    """
    转换数据
    example:

    """
    _columns_type = {"gid": int, "brand_id": int, "class_id": int, "sid": int, "country": int,
                     "price": str, "class_1_id": int, "class_2_int": int}  # "price": float,
    _columns = ["gid", "class_1_id", "class_1_name", "class_2_id", "class_2_name", "class_id", "class_name",
                "brand_id", "brand_title", "mod1", "mod", "country_title", "country", "device_name", "spare_meter",
                "meter", "device_avatar", "price", "sid",
                "address", "meter_coding", "meter_clean"]
    # 向量
    _vector_field_map = {
        "gid": "设备ID", "class_1_name": "一级分类", "class_2_name": "二级分类", "class_name": "三级分类",
        "brand_title": "品牌", "mod1": "系列", "mod": "型号", "country_title": "国别", "device_name": "设备名称",
        "address": "店铺地址", "meter_coding": "物料编码", "meter_clean": "设备属性"
    }

    _clean_client = None

    def __init__(self):
        # 连接向量数据库，通过脚本方式不会初始化项目
        rdb_conn.open()

    @staticmethod
    def format_text(texts: List[str]) -> str:
        texts = [re.sub(r"\n+\s*", ",", str(x).strip()) for x in texts if x and str(x).strip()]
        return ";".join(texts)

    @staticmethod
    def format_column(s: str | int | float):
        if type(s) == int:
            return int(s)
        if type(s) == float:
            return int(s) if int(s) == s else float(s)
        return s.strip()

    @staticmethod
    async def get_meter_clean(meter_list: List[str]) -> List[str]:
        clean = DataCleanChain()
        return await clean.abatch_clean_data(meter_list)

    async def read_excel(self, path: str) -> DataFrame | dict[IntStrT, DataFrame]:
        """
        TODO 在读取没页资源时更新meter
        :param path:
        :return:
        """
        print("----------------开始读取数据----------------\n")
        df = pd.read_excel(path, na_values="", keep_default_na=False)
        for field in df.columns.tolist():
            if field in self._columns_type:
                f_type = self._columns_type[field]
                df[field] = df[field].fillna(0).astype(f_type)
            else:
                df[field] = df[field].fillna("").astype(str)

        df.fillna("", inplace=True)
        df = df.map(lambda x: self.format_column(x))
        print("----------------结束读取数据----------------\n")

        print("----------------开始清洗数据----------------\n")
        if "meter_clean" not in df.columns:
            meter_list = df["meter"].to_list()
            meter_clean_list = await self.get_meter_clean(meter_list)
            df = df.assign(meter_clean=meter_clean_list)
        print("----------------结束清洗数据----------------\n")

        print("----------------开始组装向量数据----------------\n")
        vector_field_map = self._vector_field_map
        df["text"] = df.apply(
            lambda x: self.format_text(
                [f"{v}: {x[k]} " if k != "meter_clean" else (x[k] or x["meter"]) for k, v in vector_field_map.items()]),
            axis=1
        )
        print("----------------结束组装向量数据----------------\n")
        return df

    async def read_excel_es(self, path) -> DataFrame | dict[IntStrT, DataFrame]:
        print("----------------开始读取数据----------------\n")
        df = pd.read_excel(path, na_values="", keep_default_na=False)
        for field in df.columns.tolist():
            if field in self._columns_type:
                f_type = self._columns_type[field]
                df[field] = df[field].fillna(0).astype(f_type)
            else:
                df[field] = df[field].fillna("").astype(str)

        df.fillna("", inplace=True)
        df = df.map(lambda x: self.format_column(x))
        print("----------------结束读取数据----------------\n")
        return df

    @staticmethod
    def get_pages(total, page_size=1000):
        pages = math.ceil(total / page_size)
        return pages

    async def get_device_schema(self, df, page=1, page_size=1000):
        if df.empty:
            return [], []
        total_rows = df.shape[0]
        pages = self.get_pages(total_rows, page_size=page_size)
        if page > pages:
            return [], []
        # 计算分页的起始和结束索引
        start_index = (page - 1) * page_size
        end_index = start_index + page_size
        # 分页获取数据
        paged_data = df.iloc[start_index:end_index]
        result_list = paged_data.to_dict(orient="records")
        device_list = []
        for result in result_list:
            if not result["class_2_id"] or not str(result["class_2_id"]).isdigit():
                result["class_2_id"] = 0
            device_list.append(DeviceSchemaV2(**result))
        return device_list

